"""
Case Type   : 连接
Case Name   : 普通场景，设置idle_in_transaction_session_timeout为30s，session_timeout为0，idle状态，会话30s后不会断开
Create At   : 2024/10/12
Owner       : @li-xin12345
Description :
    1、查看enable_thread_pool
    2、创建用户
    3、设置idle_in_transaction_session_timeout为30s，session_timeout为0s；
    set idle_in_transaction_session_timeout to 30;
    set session_timeout to 0;
    show idle_in_transaction_session_timeout;
    show session_timeout;
    等待30s后查看数据库连接状态
    select now();
    4、清理环境
    gs_guc set -N all -I all -c "session_timeout = '10min'"
    gs_om -t stop && gs_om -t start
Expect      :
    1、成功
    2、成功
    3、设置成功，会话不会断开
    4、成功
History     :
"""
import os
import time
import unittest

from yat.test import Node
from yat.test import macro

from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from testcase.utils.ComThread import ComThread

pri_sh = CommonSH('PrimaryDbUser')


@unittest.skipIf(3 != pri_sh.get_node_num(), '非1+2环境不执行')
class SessionTest(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        text = f'-----{os.path.basename(__file__)} start-----'
        self.log.info(text)
        self.constant = Constant()
        self.common = Common()
        self.pri_root_node = Node('PrimaryRoot')
        self.pri_user_node = Node('PrimaryDbUser')
        self.user_name = 'u_connect_0339'
        self.script_name = 'Guc_connectionauthentication_Case0339'
        self.dest_path = os.path.join(
            macro.DB_MY_PATH, 'guc_connect_0339').replace('\\', '/')

    def test_connect(self):
        text = '----step1: 查看enable_thread_pool' \
               'expect: 显示默认值----'
        self.log.info(text)
        res = self.common.show_param('enable_thread_pool')
        self.log.info(res)
        self.assertTrue(res)
        if res != 'off':
            res = pri_sh.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                'enable_thread_pool=off')
            self.log.info(res)
            res = pri_sh.restart_db_cluster()
            self.log.info(res)

        text = '----step2: 创建用户；expect: 创建用户成功----'
        self.log.info(text)
        cmd = f"drop user if exists {self.user_name} cascade;" \
              f"create user {self.user_name} " \
              f"password '{self.pri_user_node.db_password}';" \
              f"grant all privileges to {self.user_name};"
        res = pri_sh.execut_db_sql(cmd)
        self.log.info(res)
        self.assertIn(self.constant.DROP_ROLE_SUCCESS_MSG, res,
                      f'执行失败: {text}')
        self.assertIn(self.constant.CREATE_ROLE_SUCCESS_MSG, res,
                      f'执行失败: {text}')
        self.assertIn(self.constant.ALTER_ROLE_SUCCESS_MSG, res,
                      f'执行失败: {text}')

        text = '----step3: 设置idle_in_transaction_session_timeout为30s，' \
               'session_timeout为0s；等待30s后查看数据库连接状态；' \
               'expect: 设置成功，会话不会断开----'
        self.log.info(text)
        self.log.info('---调用java脚本执行'
                      '设置idle_in_transaction_session_timeout为30s，'
                      'session_timeout为0s，及sql查询等---')
        self.log.info('--远程复制java脚本至指定主机--')
        self.common.scp_file(self.pri_root_node,
                             f"{self.script_name}.java", self.dest_path)
        cmd = f'ls {self.dest_path}'
        res = self.pri_root_node.sh(cmd).result()
        self.assertIn(self.script_name, res, '远程复制java脚本至指定主机失败')

        self.log.info('--编译java工具--')
        self.pri_root_node.scp_put(macro.JDBC_PATH,
                                   f"{self.dest_path}/postgresql.jar")
        cmd = f"javac -encoding utf-8 -cp " \
              f"{os.path.join(self.dest_path, 'postgresql.jar')} " \
              f"{os.path.join(self.dest_path, f'{self.script_name}.java')}"
        self.log.info(cmd)
        result = self.pri_root_node.sh(cmd).result()
        self.log.info(result)
        self.assertEqual('', result, '编译java工具失败')

        self.log.info('--运行java脚本--')
        url_host_port = f"{self.pri_user_node.db_host}:" \
                        f"{self.pri_user_node.db_port}"
        cmd = f" java -cp {os.path.join(self.dest_path, 'postgresql.jar')}" \
              f":{self.dest_path} " \
              f"{self.script_name} {url_host_port} " \
              f"{self.pri_user_node.db_name} {self.user_name} " \
              f"{self.pri_user_node.db_password} 2>&1"
        self.log.info(cmd)
        sql_thread = ComThread(self.pri_root_node.sh, args=(cmd,))
        sql_thread.setDaemon(True)
        sql_thread.start()
        time.sleep(2)
        query_sql = f"select * from pg_stat_activity " \
                    f"where usename = '{self.user_name}';"
        check_session_exist = pri_sh.execut_db_sql(query_sql)
        self.log.info(check_session_exist)
        self.assertIn('1 row', check_session_exist, '执行失败' + text)
        self.assertIn('idle', check_session_exist, '执行失败' + text)

        self.log.info('--等待30s，查看会话状态--')
        time.sleep(31)
        check_session_exist = pri_sh.execut_db_sql(query_sql)
        self.log.info(check_session_exist)
        self.assertIn('1 row', check_session_exist, '执行失败' + text)

        sql_thread.join(60)
        res = sql_thread.get_result()
        self.log.info(res)

    def tearDown(self):
        text = '----step4: 清理环境;expect:成功----'
        self.log.info(text)
        text1 = f'----后置操作：恢复session_timeout默认值----'
        self.log.info(text1)
        res1 = pri_sh.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG, "session_timeout='10min'")
        self.log.info(res1)
        res = pri_sh.restart_db_cluster()
        self.log.info(res)

        text2 = f'----后置操作：删除用户----'
        self.log.info(text2)
        cmd = f'drop user if exists {self.user_name} cascade;'
        res2 = pri_sh.execut_db_sql(cmd)
        self.log.info(res2)

        self.assertTrue(res1, f'执行失败: {text1}')
        self.assertIn(self.constant.DROP_ROLE_SUCCESS_MSG, res2,
                      f'执行失败: {text2}')

        text = f'-----{os.path.basename(__file__)} end-----'
        self.log.info(text)
